package connector

import (
	"bytes"
	"context"
	"fmt"
	"io/ioutil"
	"net/http"
	"sync"
	"time"
)

// HTTPSourceConnector represents an HTTP source connector
type HTTPSourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// server is the HTTP server
	server *http.Server
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// endpoint is the HTTP endpoint
	endpoint string
	// port is the HTTP port
	port int
	// method is the HTTP method
	method string
	// messages is the message channel
	messages chan *Record
	// buffer is the message buffer
	buffer []*Record
	// batchSize is the batch size
	batchSize int
}

// NewHTTPSourceConnector creates a new HTTP source connector
func NewHTTPSourceConnector() SourceConnector {
	return &HTTPSourceConnector{
		metrics:   make(map[string]interface{}),
		method:    "POST",
		messages:  make(chan *Record, 100),
		buffer:    make([]*Record, 0),
		batchSize: 10,
	}
}

// Initialize initializes the connector
func (c *HTTPSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get endpoint
	if endpoint, ok := config["endpoint"].(string); ok {
		c.endpoint = endpoint
	} else {
		c.endpoint = "/api/data"
	}

	// Get port
	if port, ok := config["port"].(float64); ok {
		c.port = int(port)
	} else {
		c.port = 8080
	}

	// Get method
	if method, ok := config["method"].(string); ok {
		c.method = method
	}

	// Get batch size
	if batchSize, ok := config["batch_size"].(float64); ok {
		c.batchSize = int(batchSize)
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *HTTPSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Create HTTP server
	mux := http.NewServeMux()
	mux.HandleFunc(c.endpoint, c.handleRequest)
	server := &http.Server{
		Addr:    fmt.Sprintf(":%d", c.port),
		Handler: mux,
	}
	c.server = server

	// Start server in a separate goroutine
	go func() {
		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			fmt.Printf("HTTP server error: %v\n", err)
		}
	}()

	c.started = true

	return nil
}

// Stop stops the connector
func (c *HTTPSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Shutdown server
	err := c.server.Shutdown(ctx)
	if err != nil {
		return fmt.Errorf("failed to shutdown HTTP server: %w", err)
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *HTTPSourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if we have buffered messages
	if len(c.buffer) > 0 {
		// Get batch size
		batchSize := c.batchSize
		if batchSize > len(c.buffer) {
			batchSize = len(c.buffer)
		}

		// Get messages
		records := c.buffer[:batchSize]
		c.buffer = c.buffer[batchSize:]

		// Update metrics
		c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
		c.metrics["last_read_time"] = time.Now().Unix()

		return NewRecordBatch(records), nil
	}

	// Read messages from channel
	for {
		select {
		case record := <-c.messages:
			// Add message to buffer
			c.buffer = append(c.buffer, record)
			// Update metrics
			c.metrics["records_total"] = c.metrics["records_total"].(int) + 1
			// Check if we have enough messages
			if len(c.buffer) >= c.batchSize {
				// Get messages
				records := c.buffer[:c.batchSize]
				c.buffer = c.buffer[c.batchSize:]

				// Update metrics
				c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
				c.metrics["last_read_time"] = time.Now().Unix()

				return NewRecordBatch(records), nil
			}
		case <-time.After(100 * time.Millisecond):
			// No message available
			if len(c.buffer) > 0 {
				// Get messages
				records := c.buffer
				c.buffer = make([]*Record, 0)

				// Update metrics
				c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
				c.metrics["last_read_time"] = time.Now().Unix()

				return NewRecordBatch(records), nil
			}
			return nil, nil
		}
	}
}

// Commit commits the offset
func (c *HTTPSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// HTTP doesn't support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *HTTPSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// handleRequest handles HTTP requests
func (c *HTTPSourceConnector) handleRequest(w http.ResponseWriter, r *http.Request) {
	// Check method
	if r.Method != c.method {
		http.Error(w, fmt.Sprintf("Method %s not allowed", r.Method), http.StatusMethodNotAllowed)
		return
	}

	// Read request body
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, fmt.Sprintf("Failed to read request body: %v", err), http.StatusBadRequest)
		return
	}

	// Create record
	record := NewRecord([]byte(r.URL.Path), body)
	record.Timestamp = time.Now()

	// Add headers
	for name, values := range r.Header {
		if len(values) > 0 {
			record.Headers[name] = values[0]
		}
	}

	// Add message to channel
	c.messages <- record

	// Send response
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("OK"))
}

// HTTPSinkConnector represents an HTTP sink connector
type HTTPSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// client is the HTTP client
	client *http.Client
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// url is the HTTP URL
	url string
	// method is the HTTP method
	method string
	// headers are the HTTP headers
	headers map[string]string
	// timeout is the HTTP timeout
	timeout time.Duration
}

// NewHTTPSinkConnector creates a new HTTP sink connector
func NewHTTPSinkConnector() SinkConnector {
	return &HTTPSinkConnector{
		metrics: make(map[string]interface{}),
		method:  "POST",
		headers: make(map[string]string),
		timeout: 30 * time.Second,
	}
}

// Initialize initializes the connector
func (c *HTTPSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get URL
	if url, ok := config["url"].(string); ok {
		c.url = url
	} else {
		return fmt.Errorf("url is required")
	}

	// Get method
	if method, ok := config["method"].(string); ok {
		c.method = method
	}

	// Get headers
	if headers, ok := config["headers"].(map[string]interface{}); ok {
		for name, value := range headers {
			if strValue, ok := value.(string); ok {
				c.headers[name] = strValue
			}
		}
	}

	// Get timeout
	if timeout, ok := config["timeout"].(float64); ok {
		c.timeout = time.Duration(timeout) * time.Second
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *HTTPSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Create HTTP client
	c.client = &http.Client{
		Timeout: c.timeout,
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *HTTPSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *HTTPSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Write records
	for _, record := range batch.Records {
		// Create request
		req, err := http.NewRequestWithContext(ctx, c.method, c.url, bytes.NewReader(record.Value))
		if err != nil {
			return fmt.Errorf("failed to create HTTP request: %w", err)
		}

		// Add headers
		for name, value := range c.headers {
			req.Header.Set(name, value)
		}
		for name, value := range record.Headers {
			req.Header.Set(name, value)
		}

		// Set content type if not set
		if req.Header.Get("Content-Type") == "" {
			req.Header.Set("Content-Type", "application/json")
		}

		// Send request
		resp, err := c.client.Do(req)
		if err != nil {
			return fmt.Errorf("failed to send HTTP request: %w", err)
		}

		// Check response
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			body, _ := ioutil.ReadAll(resp.Body)
			resp.Body.Close()
			return fmt.Errorf("HTTP request failed with status %d: %s", resp.StatusCode, string(body))
		}
		resp.Body.Close()
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *HTTPSinkConnector) Flush(ctx context.Context) error {
	// HTTP client doesn't buffer messages
	return nil
}

// GetMetrics gets connector metrics
func (c *HTTPSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}
